use smallvec::SmallVec;

use crate::{
    client::RedisClient,
    resp::{Command, Value},
    RedisError, RedisResult,
};

impl RedisClient {
    pub async fn subscribe(self, channel: &str) -> RedisResult<SubscribedStream> {
        let cmd = Command::new_with_key("SUBSCRIBE", channel);
        let (rx, socket_addr) = match &self.connection {
            crate::connection::Connection::Standalone(connection) => {
                connection.send_command_no_reply(cmd).await?;
                (connection.state.enter_stream_mode().await, None)
            }
            crate::connection::Connection::Sentinel => todo!(),
            crate::connection::Connection::Cluster(connection) => {
                let addr = connection.send_command_no_reply(cmd, None).await?;
                (connection.state.enter_stream_mode().await, Some(addr))
            }
        };
        Ok(SubscribedStream::new(
            self,
            smallvec::smallvec![channel.to_string()],
            rx,
            socket_addr,
        ))
    }

    pub async fn subscribe_more(self, channels: &[&str]) -> RedisResult<SubscribedStream> {
        let cmd = Command::new_with_keys("SUBSCRIBE", channels);
        let mut sub_channels = smallvec::smallvec![];
        for channel in channels {
            sub_channels.push(channel.to_string());
        }
        let (rx, socket_addr) = match &self.connection {
            crate::connection::Connection::Standalone(connection) => {
                connection.send_command_no_reply(cmd).await?;
                (connection.state.enter_stream_mode().await, None)
            }
            crate::connection::Connection::Cluster(connection) => {
                let addr = connection.send_command_no_reply(cmd, None).await?;
                (connection.state.enter_stream_mode().await, Some(addr))
            }
            crate::connection::Connection::Sentinel => todo!(),
        };
        Ok(SubscribedStream::new(self, sub_channels, rx, socket_addr))
    }

    // async fn send_command_no_reply<'a>(&self, cmd: Command<'a>) -> RedisResult<()> {
    //     match &self.connection {
    //         crate::connection::Connection::Standalone(connection) => {
    //             connection.send_command_no_reply(cmd).await
    //         }
    //         crate::connection::Connection::Sentinel => todo!(),
    //         crate::connection::Connection::Cluster => todo!(),
    //     }
    // }
}

pub struct SubscribedMessage {
    pub channel: String,
    pub content: Vec<u8>,
}

impl SubscribedMessage {}

pub struct SubscribedStream {
    client: RedisClient,
    channels: SmallVec<[String; 4]>,
    socket_addr: Option<String>,
    rx: tokio::sync::mpsc::Receiver<Value>,
}

impl SubscribedStream {
    fn new(
        client: RedisClient,
        channels: SmallVec<[String; 4]>,
        rx: tokio::sync::mpsc::Receiver<Value>,
        socket_addr: Option<String>,
    ) -> Self {
        Self {
            client,
            channels,
            rx,
            socket_addr,
        }
    }

    pub async fn recv(&mut self) -> RedisResult<Option<SubscribedMessage>> {
        loop {
            let value = self.rx.recv().await;
            match value {
                Some(Value::Pushes(mut array)) => {
                    if array.len() < 2 {
                        return Err(RedisError::WaitReceive(
                            "the len of subscribed result is invalid".to_string(),
                        ));
                    }
                    let t = array.remove(0).into_string()?;
                    // subscribe or unsubscribe or message
                    if t == "message" {
                        if array.len() < 2 {
                            return Err(RedisError::WaitReceive(
                                "the len of subscribed message result is invalid".to_string(),
                            ));
                        }
                        return Ok(Some(SubscribedMessage {
                            channel: array.remove(0).into_string()?,
                            content: array.remove(0).into_vec_u8()?,
                        }));
                    }
                }
                None => {
                    return Ok(None);
                }
                _ => {
                    return Err(RedisError::WaitReceive(
                        "the type of subscribed result is invalid".to_string(),
                    ));
                }
            }
        }

        // match self.rx.recv().await {
        //     Some(value) => match value {
        //         Value::Pushes(array) => {

        //         }
        //         _ => {
        //             return Err(RedisError::Send(
        //                 "the type of subscribed result is invalid".to_string(),
        //             ));
        //         }
        //     },
        //     None => {

        //     }
        // }
        // self.rx.recv().await
    }

    pub async fn exit(mut self) -> RedisResult<RedisClient> {
        let temp = self
            .channels
            .iter()
            .map(|x| x.as_str())
            .collect::<Vec<&str>>();
        let cmd = Command::new_with_keys("UNSUBSCRIBE", temp.as_slice());
        match &self.client.connection {
            crate::connection::Connection::Standalone(connection) => {
                connection.send_command_no_reply(cmd).await?;
                match self.rx.recv().await {
                    Some(_) => {
                        // log::info!("unsubscribe val = {val:?}");
                    }
                    None => {}
                }
                connection.state.exit_stream_mode().await;
            }
            crate::connection::Connection::Cluster(connection) => {
                connection
                    .send_command_no_reply(cmd, self.socket_addr)
                    .await?;
                match self.rx.recv().await {
                    Some(_) => {
                        // log::info!("unsubscribe val = {val:?}");
                    }
                    None => {}
                }
                connection.state.exit_stream_mode().await;
            }
            crate::connection::Connection::Sentinel => todo!(),
        };
        self.rx.close();
        Ok(self.client)
    }
}
